package hn.cch.custom;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Headers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

public class CustomProducerInterceptor implements ProducerInterceptor<String, String> {

    private static final Logger logger = LoggerFactory.getLogger(CustomProducerInterceptor.class);


    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
        String key = producerRecord.key();
        String value = producerRecord.value();
        String topic = producerRecord.topic();
        Integer partition = producerRecord.partition();
        Long timestamp = producerRecord.timestamp();
        Headers headers = producerRecord.headers();
        logger.info("{},{},{},{},{}", topic, partition, timestamp, key, value);
        return new ProducerRecord<>(topic, partition, timestamp, key, value, headers);
    }

    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {

    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}
